<?php

namespace App\Process;

use App\Model\Play;
use App\Utils\Enum;
use App\Utils\Msg;
use App\Websocket\ScoreWebsocket;
use Exception;
use Hyperf\Coroutine\Coroutine;
use Hyperf\Di\Annotation\Inject;
use Hyperf\Logger\LoggerFactory;
use Hyperf\Process\AbstractProcess;
use Hyperf\Process\Annotation\Process;
use Hyperf\Redis\Redis;
use Hyperf\WebSocketClient\Client;
use Hyperf\WebSocketClient\ClientFactory;
use Hyperf\WebSocketServer\Sender;
use Psr\Container\ContainerExceptionInterface;
use Psr\Container\ContainerInterface;
use Psr\Container\NotFoundExceptionInterface;
use Psr\Log\LoggerInterface;
use Swoole\Timer;
use function Hyperf\Support\env;

#[Process(name: 'live-match-transfer')]
class MatchTransferProcess extends AbstractProcess
{
    /**
     * 进程数量
     */
    public int $nums = 1;

    /**
     * 进程名称
     */
    public string $name = 'live-match-transfer';

    /**
     * 重定向自定义进程的标准输入和输出
     */
    public bool $redirectStdinStdout = false;

    /**
     * 管道类型
     */
    public int $pipeType = 2;

    /**
     * 是否启用协程
     */
    public bool $enableCoroutine = true;
    /**
     * @var Sender
     */
    #[Inject]
    protected Sender $sender;

    protected LoggerInterface $logger;

    protected Client $client;
    #[Inject]
    protected ClientFactory $clientFactory;

    protected int $heartbeatTimerId = 0;
    protected int $healthCheckTimerId = 0;
    protected int $lastReceiveHeartbeatTime = 0;
    protected int $lastReceiveBusinessMsgTime = 0;
    /**
     * 连接最大空闲时间
     * @var int
     */
    protected int $maxIdleTime = 0;

    public function __construct(ContainerInterface $container)
    {
        parent::__construct($container);
        try {
            $this->logger = $this->container->get(LoggerFactory::class)->get('live-match-transfer', 'process');
            $this->maxIdleTime = (int)env('TARGET_WS_MAX_IDLE', 30);
        } catch (ContainerExceptionInterface|NotFoundExceptionInterface $e) {
        }
    }

    private function exit(): void
    {
        $this->client->close();
        Timer::clearAll();
        foreach (\Swoole\Coroutine::list() as $cid) {
            \Swoole\Coroutine::cancel($cid);
        }
        $this->process->exit();
    }

    public function handle(): void
    {
        $this->init();

        $this->lastReceiveBusinessMsgTime = time();
        $this->healthCheckTimerId = Timer::tick(3000, function () {
            $this->logger->info(sprintf('pid=%d 检查连接是否健康...', $this->process->pid));
            if ($this->lastReceiveHeartbeatTime && time() - $this->lastReceiveHeartbeatTime > 25) {
                $this->logger->warning('websocket心跳返回超时，即将重连');
                $this->exit();
            }
            //30s内未收到重要的消息，则可能ws服务器方出问题了，需要重连
            if ($this->maxIdleTime > 0 && time() - $this->lastReceiveBusinessMsgTime > $this->maxIdleTime) {
                $this->logger->warning(sprintf('连接空闲时间超过%d秒，需要断线重连', $this->maxIdleTime));
                $this->exit();
            }
        });
        while (true) {
            $receive = $this->client->recv();
            if ($receive === false) {
                break;
            }
            try {
                $msg = Msg::Parse($receive);
                if ($msg) {
                    $this->logger->info(sprintf('收到%s消息', $msg['msgType']));
                    $action = '_event' . ucfirst($msg['msgType']);
                    if (method_exists($this, $action)) {
                        go(function () use ($action, $msg) {
                            $this->{$action}($msg['msgBody']);
                        });
                    } else {
                        $this->logger->warning("未知消息类型:{$msg['msgType']}");
                    }
                } else {
                    $this->logger->warning('消息解析失败', [$receive]);
                }
            } catch (Exception $e) {
                $this->logger->warning(sprintf('消息解析失败:%s', $e->getMessage()), [$receive]);
            }
        }
        $this->logger->warning('接收消息失败,链接或已断开，进程即将退出');
        $this->exit();
    }


    private function init(): void
    {
        $this->logger->info('初始化...');
        if ($this->heartbeatTimerId) {
            Timer::clear($this->heartbeatTimerId);
        }
        if ($this->healthCheckTimerId) {
            Timer::clear($this->healthCheckTimerId);
        }
        try {
            $this->client = $this->clientFactory->create(env('TARGET_WS_ADDRESS', ''), false, [
                'UserAgent' => Enum::HTTP_COMMON_HEADERS['User-Agent'],
            ]);

            //登录
            $this->logger->info('UserEntryCmd');
            $this->client->push(
                Msg::Create('UserEntryCmd', [
                    'DeviceId' => md5((string)time()),
                ])
            );

            $heartbeatMsg = Msg::Create('UserHeartbeatCmd');
            $this->heartbeatTimerId = Timer::tick(10000, function () use ($heartbeatMsg) {
                $ret = $this->client->push($heartbeatMsg, SWOOLE_WEBSOCKET_OPCODE_BINARY);
                $this->logger->info('发送心跳:' . ($ret ? '成功' : '失败'));
            });
        } catch (Exception $e) {
            $this->logger->warning('链接异常:' . $e->getMessage());
            Coroutine::sleep(3);
            Timer::clear($this->heartbeatTimerId);
            $this->heartbeatTimerId = 0;
            $this->init();
        }
    }

    /**
     * 心跳返回
     * @param $msg
     * @return void
     */
    protected function _eventUserHeartbeatCmd($msg): void
    {
        $this->lastReceiveHeartbeatTime = time();
        $this->logger->info('收到心跳返回');
    }

    /**
     * UserDoResult
     * @param $msg
     * @return false|void
     */
    protected function _eventUserDoResult($msg)
    {
        $this->lastReceiveBusinessMsgTime = time();
        $data = new_json_decode($msg['data']);
        if (!$data || !isset($data['topic']) || !preg_match('#^/sports/\d+/match/scores$#', $data['topic'])) {
            $this->logger->warning('消息内容不符合广播:' . $msg['data']);
            return false;
        }
        $dataList = [];
        $this->logger->info(sprintf('收到%d条比分更新', count($data['data']['scores'])));
        foreach ($data['data']['scores'] as $rs) {
            $match = Play::query()->where('match_id', $rs['matchId'])->first();
            if (!$match) {
                $this->logger->warning("比赛(matchId={$rs['matchId']})在数据库中未找到");
                continue;
            }

            $update = [
                'match_status' => $rs['matchStatus'],
                'match_status_id' => $rs['statusId'],
            ];
            if ($match['match_status'] != Enum\MatchEnum::MATCH_STATUS_END && $rs['matchStatus'] == Enum\MatchEnum::MATCH_STATUS_END) {
                $update['match_end_time'] = time();
            }
            if (isset($rs['homeScores'])) {
                $update['home_team_scores'] = new_json_encode($rs['homeScores']);
                $update['away_team_scores'] = new_json_encode($rs['awayScores']);
            }
            if (isset($rs['times'])) {
                $update['times'] = new_json_encode($rs['times']);
            }
            if (isset($rs['tennisGamePoints'])) {
                $update['tennis_game_points'] = new_json_encode($rs['tennisGamePoints']);
            }
            if (isset($rs['vbScores'])) {
                $update['vb_scores'] = new_json_encode($rs['vbScores']);
            }
            if (isset($rs['serve'])) {
                $update['tennis_serve'] = $rs['serve'];
            }
            Play::query()->where('id', $match['id'])->update($update);
            $dataList[] = $rs;
        }
        $this->logger->info(sprintf('可推送数据%d条', count($dataList)));
        if (!empty($dataList)) {
            $packMsg = Msg::Create('UserDoResult', [
                'data' => new_json_encode($data['data']),
            ]);
            try {
                $redis = $this->container->get(Redis::class);
                $succ = 0;
                $fdList = $redis->sMembers(ScoreWebsocket::KEY_FDS);
                if (is_array($fdList) && !empty($fdList)) {
                    $total = count($fdList);
                    foreach ($fdList as $fd) {
                        $ret = $this->sender->push((int)$fd, $packMsg, SWOOLE_WEBSOCKET_OPCODE_BINARY);
                        if ($ret) {
                            ++$succ;
                        }
                    }
                    $this->logger->info(sprintf('成功向 %d/%d 客户端推送', $succ, $total));
                } else {
                    $this->logger->info('客户端数量=0 无需推送');
                }
            } catch (ContainerExceptionInterface|NotFoundExceptionInterface|\RedisException $e) {
                $this->logger->warning('sender异常:' . $e->getMessage());
            }
        }
    }
}